package producer.broker;

import com.api.Exception.MessageRunTimeException;
import com.api.Message;
import com.api.MessageType;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.units.qual.A;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.converter.GenericMessageConverter;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import producer.service.MessageStoreService;

import java.util.List;
import java.util.Map;

/**
 * $RabbitTemplateContainer mq链接池 封装
 * 1.每一个topic 对应一个链接
 * 2. 可以提高效率
 * 3. 更具msg定制一个链接
 * Created with IDEA
 * Author: LuoHengYi
 * Date: 2022/6/18
 * Time: 15:04
 */
@Slf4j
@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback{

    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private MessageStoreService messageStoreService;

    private Map<String /* TOPIC 也就是交换机名称 */, RabbitTemplate> rabbitMap =   Maps.newConcurrentMap();

    private Splitter splitter = Splitter.on("#");

    public RabbitTemplate getRabbitTemplate(Message message) throws MessageRunTimeException {
        Preconditions.checkNotNull(message);

        String topic = message.getTopic();


        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);

        if (rabbitTemplate !=null){
            return  rabbitTemplate;
        }

        log.info("#RabbitTemplateContainer.getRabbitTemplate# topic：{} is not exit, create one",topic);

        RabbitTemplate newRabbitTemplate = new RabbitTemplate(connectionFactory);
        //设置交换机
        newRabbitTemplate.setExchange(topic);
        //设置RoutingKey
        newRabbitTemplate.setRoutingKey(message.getRoutingKey());
        newRabbitTemplate.setRetryTemplate(new RetryTemplate());

        // 对于msg的序列化 可以自己去实现一个 ConversionService
        //GenericMessageConverter gmc = new GenericMessageConverter();
        //newRabbitTemplate.setMessageConverter();
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

        //只要不是快速消息都需要消息确认
        String messageType = message.getMessageType();
        if (!MessageType.RAPID.equals(messageType)){
            newRabbitTemplate.setConfirmCallback(this);
        }
        rabbitMap.putIfAbsent(topic,newRabbitTemplate);

        return newRabbitTemplate;
    }


    //消息发送成功后，rabbitMQ对于客户端的应答
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //  这里的 correlationData 是发送消息时，携带发送的 CorrelationData 对象 @RabbitBrokerImpl.sendKernel 中的correlationData对象

        // splitter 将发送时携带的数据按#分割成list
        List<String> list = splitter.splitToList(correlationData.getId());

        String messageId = list.get(0);
        long time = Long.parseLong(list.get(1));

        //如果发送成功！
        if (ack){
            log.info(" send msg is ok，confirm msgId:{},sendTime:{}",messageId,time);
            //修改数据库log记录
            messageStoreService.success(messageId);

        }else {
            log.info(" send msg is error，confirm msgId:{},sendTime:{}",messageId,time);
        }


    }
}
